package com.etlmaster.executor;

import java.util.Stack;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import com.etlmaster.executor.StepStack.MetaEntry;
import com.etlmaster.executor.steps.control.ConditionalStepInterface;
import com.etlmaster.executor.steps.control.LoopStepInterface;
import com.etlmaster.executor.utils.CallableResult;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.DpExecutorDbLogUtils;
import com.etlmaster.executor.utils.LogWriter;
import com.etlmaster.executor.utils.StepState;

/**@author Larry
 * @date 2017-04-14
 * 流程执行器
 */
public class EtlMasterExecutor {

	private static Stack<StepMeta> stack = new Stack<StepMeta>();

	public boolean executeByStack(Job job) {
		LogWriter.addLog(job.getMeta().getId(),0,"INFO",
				"------------------{}开始执行----------------------", job.getMeta()
						.getId());
		String jobId = job.getMeta().getId();
		String jobBatchNo = job.getVariable("taskid") != null ? job
				.getVariable("taskid").toString() : "19000101";
		String user = job.getVariable("user") != null ? job.getVariable("user")
				.toString() : null;
		boolean isIgnore=Boolean.parseBoolean( (String) job.getVariable("ignore"));
		if(!isIgnore)
		DpExecutorDbLogUtils.logProcStart(job.getSeesionId(), jobId,
				jobBatchNo, user);

		StepStack stack =StepStack.stepStacks.get(jobId);
		if(stack==null)
			stack= new StepStack(jobId+"_"+job.getSeesionId(), DataCollectionUtils.getIntProperties("defaultStackSize"), false);
		stack.pushStack(stack.getHelper().getWaitQueue(), job.getRootStepMetas(), job,job.getSeesionId());
		
		try {
			stack.setAllFetched(true);
			Future<CallableResult> future=stack.getHelper().executor.submit(stack.getStackThread());
			
			CallableResult result=future.get();
			if(result.isError()){
				throw  new EtlMasterException(result.getMessage());
			}
		} catch (InterruptedException e) {
			throw  new EtlMasterException(e.getMessage());

		} catch (ExecutionException e) {
			throw  new EtlMasterException(e.getMessage());
		}
		
		if(!isIgnore)
		DpExecutorDbLogUtils.logProcEnd(job.getSeesionId(), jobId, jobBatchNo,
				"0", "", user);
		LogWriter.addLog(job.getMeta().getId(),0,"INFO",
				"------------------{}执行结束----------------------", job.getMeta()
						.getId());
		jobSuccessed(job);// TODO 需要更为严谨的验证，如验证是否正常轨迹，是否每个step都成功等
		return job.getSuccessed();
	}

	private void jobSuccessed(Job job) {
		job.setSuccessed(true);
	}

	public static void abort() {
		/*
		 * job.setSuccessed(false); List<Step> runningSteps = getRunningStep();
		 * for(Step step : runningSteps){ step.stop(); step.setSuccessed(false);
		 * } LOG.error("步骤执行失败，程序执行中止");
		 */
	}

	public static void stop(Job job) {
		// job.setSuccessed(false);
		System.exit(0);
		// throw new DpExecutorException("步骤未执行，程序停止");
	}

	public void suspend() {

	}

	public void resume() {

	}

	public static void pushStack(StepMeta stepMeta) {
		if (stepMeta != null) {
			stack.push(stepMeta);
		}
	}
	public static void runStep(StepStack stack, MetaEntry entry) {
		if (entry.getMeta() != null) {

			entry.getJob().getVariables().put("finishStepId", entry.getMeta().getId());

			DependencyCheck.debugMode(entry.getMeta(), entry.getJob());
			if (entry.getMeta() instanceof ConditionalStepInterface) {

				entry.setMeta(((ConditionalStepInterface) entry.getMeta()).process(
						entry.getMeta(), entry.getJob()));
			}
			DependencyCheck.debugMode(entry.getMeta(), entry.getJob());
			if (entry.getMeta() instanceof LoopStepInterface) {

				((LoopStepInterface) entry.getMeta()).loop(stack, entry.getMeta(),
						entry.getJob());
				if(entry.getMeta().getStepState().getStatus()!=StepState.RUNNING_SUCCESSED){
					throw new EtlMasterException("循环执行失败");
				}
			} else {

				handleStep(stack, entry.getMeta(), entry.getJob());
			}
		} else {
			LogWriter.addLog(entry.getJob().getMeta().getId(),0,"ERROR", "当前步骤信息空");
		}
	}

	/**
	 * TODO 需要重构使用消息发送事件
	 * 
	 * @param stepMeta
	 * @param job
	 */
	public static void handleStep(StepStack stack, StepMeta stepMeta, Job job) {
		Step step = ((Step) stepMeta.getStep());
		LogWriter.addLog(job.getMeta().getId(),0,"INFO", "开始执行[步骤Id:{}({})]", step.getMeta().getId(),
				step.getMeta().getName());
		try {
			step.run(job.getMeta());
			
		} catch (Exception e) {
			LogWriter.addLog(job.getMeta().getId(),0,"ERROR", "步骤[{}]运行失败堆栈信息,{}", stepMeta.getName(),
					e.getMessage());
			jobFailed(job);
		}
	}

	private static void jobFailed(Job job) {
		abort();
		job.setSuccessed(false);
		throw new EtlMasterException("步骤{}未执行失败，程序退出");
	}

	public static void main(String[] args) {

	}

	public static void error(Job job) {
		LogWriter.addLog(job.getMeta().getId(),0,"INFO",
				"------------------{}执行结束----------------------", job.getMeta()
						.getId());
		LogWriter.addLog(job.getMeta().getId(),0,"ERROR", "程序执行失败，程序名:{} 失败的步骤id为：{}", job.getMeta()
				.getName(), job.getVariable("finishStepId") + "!");

		DpExecutorDbLogUtils.logProcEnd(
				job.getSeesionId(),
				job.getMeta().getId(),
				job.getVariable("taskid") != null ? job.getVariable("taskid")
						.toString() : "19000101",
				(String) job.getVariable("finishStepId"),
				"运行失败",
				job.getVariable("user") != null ? (String) job
						.getVariable("user") : null);
		System.exit(Integer.parseInt((String) job.getVariable("finishStepId")));

	}
}
